package com.margo.rabbit.topic;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * topic多个生产者，一个消费者的情况
 * @author fangxingbang
 *
 */
public class MargoRabbitTopicConsumer {
	
	public static void main(String[] args) throws IOException, TimeoutException {
		topicConsumer();
	}
	/**
	 * topic模式
	 * @throws IOException
	 * @throws TimeoutException
	 */
	public static void topicConsumer() throws IOException, TimeoutException {
		//rabbit重启以后，交换器与队列消失
		String exchange="margo-topic";
		String queue="margo-topic-queue";
		String routingKey="margo-#";
		ConnectionFactory connectionFactory=new ConnectionFactory();
		connectionFactory.setHost("127.0.0.1");
		Connection connection=connectionFactory.newConnection();
		Channel channel=connection.createChannel();
		//申明交换器
		channel.exchangeDeclare(exchange, BuiltinExchangeType.TOPIC);
		//申明队列
		channel.queueDeclare(queue, false, false, false, null);
		//交换器与队列 路由键绑定
		channel.queueBind(queue, exchange, routingKey);
		channel.basicConsume(queue, new DefaultConsumer(channel) {
			 @Override
	         public void handleDelivery(String consumerTag,
	                                       Envelope envelope,
	                                       AMQP.BasicProperties properties,
	                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println("Received routingKey["
                        +envelope.getRoutingKey()
                        +"]"+message);
			 }
		});
	}
}
